package com.doitedu.core

import com.doitedu.beans.UserBean
import com.doitedu.utils.SparkUtil
import org.apache.spark.rdd.JdbcRDD

import java.sql.{Connection, DriverManager, ResultSet}

/**
 * @Date: 22.6.29 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 * 连接关系型数据库  创建RDD
 */
object C07MakeRDD_JDBC {

  val  getConnection = ()=>{
   val conn: Connection =  DriverManager.getConnection("jdbc:mysql://localhost:3306/cool?characterEncoding=UTF8" ,"root","root")
    conn
  }
  /**
   * 处理每行结果数据
   */
  val mapRow= (rs:ResultSet)=>{ // 封装一行数据
    val id = rs.getInt(1)
    val account = rs.getString(2)
    val password = rs.getString(3)
    val memberLevel = rs.getString(4)
    val phonenbr = rs.getString(5)
    UserBean(id , account, password , memberLevel , phonenbr)
  }

  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSparkContext("mk_RDD_mysql")

    /**
     * 使用new  JdbcRDD 是RDD的一个子类
     * 泛型  返回封装的数据类型
     * 参数1 :  Spark的环境
     * 参数2 :  MySQL的连接对象
     * 参数3 :  执行的sql语句  查询
     *    select  * from user where id >= ?  and  id <= ?
     * 参数4 :  查询数据的范围  第一个问号
     * 参数5 :  查询数据的范围  第二个问号
     * 参数6 :  分区个数
     * 参数7 :   封装查询结果数据的逻辑
     */
    val rdd1: JdbcRDD[UserBean] = new JdbcRDD[UserBean](
      sc,
      getConnection,
      "select  * from user where id >= ?  and  id <= ?",
      1, 3,
      1,
      mapRow
    )

    rdd1.foreach(println)
    sc.stop

  }

}
